package cn.simon.free.redis.mq.config;

import cn.simon.free.redis.mq.annotations.RedisMessageConsumer;
import cn.simon.free.redis.mq.config.properties.RedisMessageProperties;
import cn.simon.free.redis.mq.core.consumer.AbstractRedisMessageConsumer;
import cn.simon.free.redis.mq.core.consumer.DelayRedisMessageConsumer;
import cn.simon.free.redis.mq.core.consumer.NormalRedisMessageConsumer;
import cn.simon.free.redis.mq.core.consumer.task.ThreadPoolLongPollingTask;
import cn.simon.free.redis.mq.enums.ConsumerType;
import cn.simon.free.redis.mq.exceptions.MessageTypeException;
import com.google.common.collect.Sets;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeansException;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.event.EventListener;

import java.util.HashSet;
import java.util.Map;
import java.util.Optional;

/**
 * @author simon
 * @date 2021/7/1 14:50
 * @copyright:
 * @modifier
 * @motto a no talent, people who have to study hard
 */
@Slf4j
@Configuration
@ConditionalOnBean(RedisMessageInitConfig.class)
public class RedisMessageConsumerInitConfig implements ApplicationContextAware {

    private final HashSet<String> queueSets = Sets.newHashSet();
    private ApplicationContext applicationContext;
    private RedisMessageProperties redisMessageProperties;

    /**
     * spring 启动事件之初始化步骤都将完成 {@link org.springframework.boot.context.event.ApplicationReadyEvent}
     */
    @EventListener({ApplicationReadyEvent.class})
    public void start() {
        Map<String, Object> beans = applicationContext.getBeansWithAnnotation(RedisMessageConsumer.class);
        beans.forEach((key, object) -> {
            RedisMessageConsumer redisMessageConsumer = Optional.ofNullable(applicationContext.findAnnotationOnBean(key, RedisMessageConsumer.class))
                    .orElseThrow(() -> new MessageTypeException("spring初始化完成,但[" + key + "]不存在与Spring上下文中,请检查"));
            String queueName = redisMessageConsumer.queueName();
            ConsumerType consumerType = redisMessageConsumer.consumerType();
            //快速去重
            if (queueSets.contains(queueName)) {
                throw new MessageTypeException("队列名:" + queueName + "检测到重名,请检查.");
            }
            queueSets.add(queueName);
            //配置初始化
            redisMessageProperties.getQueueNameWithHash().put(queueName, redisMessageConsumer);
            redisMessageProperties.getQueueNameWithBeanHash().put(queueName, object);
            //选取核心模式
            ThreadPoolLongPollingTask threadPoolLongPollingTask = redisMessageConsumer.singleThreadProcess() ?
                    new ThreadPoolLongPollingTask(1, queueName, consumerType, (AbstractRedisMessageConsumer) object) :
                    new ThreadPoolLongPollingTask(redisMessageConsumer.corePoolSize(), queueName, consumerType, (AbstractRedisMessageConsumer) object);

            if (DelayRedisMessageConsumer.class.isAssignableFrom(object.getClass())
                    || NormalRedisMessageConsumer.class.isAssignableFrom(object.getClass())) {
                threadPoolLongPollingTask.start();
                return;
            }
            log.warn(String.format("检测到消息队列:%s,为暂不支持的消息类型:%s", queueName, redisMessageConsumer.messageType()));
        });
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
        this.redisMessageProperties = applicationContext.getBean(RedisMessageProperties.class);
    }
}
